package net.bwie.realtime.jtp.dwd.trade.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.dwd.trade.function.LoadProvinceDimMapFunction;
import net.bwie.realtime.jtp.dwd.trade.function.RequestDicDimAsyncFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class JtpOrderDetailDorisDwdJob {
    public static void main(String[] args) throws Exception {
        //1.  创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2. 读取kafka数据
        DataStream<String> dataStream = KafkaUtil.consumerKafka(env, "topic-db");
//        dataStream.print();
        //3.处理数据
        DataStream<String> dwdStream= handle(dataStream);
//        dwdStream.print();
        env.execute();

    }

    private static DataStream<String> handle(DataStream<String> stream) {
        //订单明细数据流
        DataStream<String> joinStream = joinProcess(stream);
        joinStream.print("join");
        //订单信息数据流
        SingleOutputStreamOperator<String> province = joinStream.map(new LoadProvinceDimMapFunction());
//        province.print("province");
        //异步维表关联
//        SingleOutputStreamOperator<String> dwdStrram = AsyncDataStream.orderedWait(province, new RequestDicDimAsyncFunction(),
//                10000, TimeUnit.MILLISECONDS, 100);
//        return dwdStrram;
        return null;
    }

    private static DataStream<String> joinProcess(DataStream<String> stream) {
        SingleOutputStreamOperator<String> detailStream = stream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);

                return "order_detail".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });
        SingleOutputStreamOperator<String> infoStream = stream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return "order_info".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });
        SingleOutputStreamOperator<String> timeInfoStream = infoStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                String eventTime = JSON.parseObject(element)
                                        .getJSONObject("operate_date").getString("create_time");

                                return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                            }
                        })
        );
        SingleOutputStreamOperator<String> timeDetailStream = detailStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.
                        <String>forBoundedOutOfOrderness(
                        Duration.ofSeconds(0)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {
                        String eventTime = JSON.parseObject(element)
                                .getJSONObject("operate_date")
                                .getString("create_time");
                        //  转换时间格式
                        return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                    }
                })
        );
        /**
         * 数据流关联Join，由于info -> detail ，所以使用Interval join
         * interval join 要求2个数据流为分组流（keyed stream）：按照关联字段分组keyBy
         */

        SingleOutputStreamOperator<String> joinStream = timeInfoStream
                .keyBy(element -> JSON.parseObject(element)
                        .getJSONObject("operate_date").getString("id")
                ).intervalJoin(
                        timeDetailStream.keyBy(element -> JSON.parseObject(element)
                                .getJSONObject("operate_date").getString("order_id")
                        )
                ).between(Time.seconds(-1), Time.seconds(2))
                .process(new ProcessJoinFunction<String, String, String>() {
                    @Override
                    public void processElement(String info, String detail,
                                               Context ctx, Collector<String> out) throws Exception {
                        JSONObject jsonobject = JSON.parseObject(info).getJSONObject("operate_date");
                        Integer userId = jsonobject.getInteger("user_id");
                        Integer provinceId = jsonobject.getInteger("province_id");
                        JSONObject detailJsonObject = JSON.parseObject(detail).getJSONObject("operate_date");
                        detailJsonObject.put("user_id", userId);
                        detailJsonObject.put("province_id", provinceId);
                        String createTime = detailJsonObject.getString("create_time");
                        String curDate = createTime.substring(0, 10);
                        detailJsonObject.put("cur_date", curDate);
                        String output = detailJsonObject.toJSONString();
                        out.collect(output);
                    }
                });
        return joinStream;
    }
}
